package com.yjd.bi.etl.canal;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

import com.alibaba.otter.canal.protocol.CanalEntry.EventType;
import com.yjd.bi.canal.model.CanalMessageModel;
import com.yjd.bi.canal.service.ICanalMessageService;
import com.yjd.comm.annotation.ServiceRemoteProxy;
import com.yjd.comm.base.enums.DbWREnums;
import com.yjd.job.BaseJob;
import com.yjd.comm.rpm.RpmServiceKeyEnum;
import com.yjd.comm.util.FrameUtil;

public class CanalDataTransferJob extends BaseJob{

	private static int GEN_BANCH=500;
	@ServiceRemoteProxy(serviceUrlResouceKey = RpmServiceKeyEnum.RPM_BI_SERVICE_KEY)
	ICanalMessageService canalMessageService;
	
	@Override
	protected boolean doExecute() throws Exception {
		return transfer();
	}

	@Override
	protected String getJobType() {
		return CanalDataTransferJob.class.getName();
	}

	/**
	 * 
	 * @return
	 * @throws Exception 
	 */
	@SuppressWarnings({ "unchecked", "rawtypes" })
	public boolean transfer() throws Exception{
		Map paramMap=FrameUtil.newHashMap("status",0,"event_type_min",EventType.INSERT.getNumber(),"event_type_max",EventType.ALTER.getNumber(),"dbtype","mysql","rowstartindex",0,"rowendindex", GEN_BANCH);
		List<CanalMessageModel> list=canalMessageService.getModelList(paramMap,DbWREnums.WRITE);
		int sum=list.size();
		while (sum!=0){
			canalMessageService.transfer(list);
			list=canalMessageService.getModelList(paramMap,DbWREnums.WRITE);
			paramMap=FrameUtil.newHashMap("status",0,"event_type_min",EventType.INSERT.getNumber(),"event_type_max",EventType.ALTER.getNumber(),"dbtype","mysql","rowstartindex",0,"rowendindex", GEN_BANCH);
			sum=list.size();
		}
		return true;
	}
	
	
	
}
